package com.kc.gb.device.plugins.strip.service.impl;

import cn.hutool.core.net.NetUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.kc.gb.device.common.adapter.DeviceAdaptor;
import com.kc.gb.device.common.contants.MQConst;
import com.kc.gb.device.common.dto.I18NException;
import com.kc.gb.device.common.enums.ErrorCode;
import com.kc.gb.device.common.properties.DeviceProperties;
import com.kc.gb.device.plugins.strip.service.StripHardwareService;
import com.kc.gb.device.plugins.strip.service.StripMQService;
import com.kc.gb.feign.strip.sopport.dto.StripCommand;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
@Service
public class StripMQServiceImpl implements StripMQService {

    @Resource
    DeviceProperties deviceProperties;

    @Resource
    DeviceAdaptor deviceAdaptor;

    @Resource
    StripHardwareService iStripHardwareService;

    private static DefaultMQPushConsumer ptlCommandConsumer;

    @Override
    public DefaultMQPushConsumer createCommandConsumer() throws Exception {
        String station = deviceAdaptor.getApplicationStationCode();
        if(StringUtils.isBlank(station)){
            throw new I18NException(ErrorCode.STATION_NOT_BIND);
        }
        String topic = MQConst.Strip.strip_station_to_device_topic;
        String consumerGroup = MQConst.Strip.strip_station_to_device_topic_consumer;
        String nameServer= getNameServer();
        // Strip 消费逻辑
        MessageListenerOrderly listener = new MessageOrderlyConsumer();
        String tag = station;
        DefaultMQPushConsumer consumer = registerConsumer(nameServer,topic,consumerGroup,tag,listener);
        ptlCommandConsumer = consumer;
        return consumer;
    }

    public DefaultMQPushConsumer registerConsumer(String namesrv, String topic, String consumerGroup, String tagsFilter, MessageListenerOrderly listener) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        // 开启消息轨迹 ConsumeMessageTraceHookImpl
        // DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup,true);
        // consumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(new SleuthMessageConsumeHook());
        consumer.setNamesrvAddr(namesrv);
        //一个GroupName第一次消费时的位置
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 消费者线程池 ThreadPoolExecutor 线程池大小控制；ConsumeThreadMin(corePoolSize),ConsumeThreadMax(maximumPoolSize)
        consumer.setConsumeThreadMin(20);
        consumer.setConsumeThreadMax(20);
        //要消费的topic，可使用tag进行简单过滤
        consumer.subscribe(topic, tagsFilter);
        //PullBatchSize指的是一次从broker的一个消息队列获取消息的最大数，默认是32
        consumer.setPullBatchSize(32);
        //consumer.subscribe("TestTopic", MessageSelector.bySql("(status is not null and status>1 )"));
        //单次批量消费的数量，默认1
        consumer.setConsumeMessageBatchMaxSize(1);
        //消费模式，广播或者集群，默认集群。
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //在同一jvm中 需要启动两个同一GroupName的情况需要这个参数不一样。
        String instanceName = NetUtil.getLocalhostStr();
        consumer.setInstanceName(instanceName);
        //配置消息监听
        consumer.registerMessageListener(listener);
        consumer.start();
        return consumer;
    }

    public class MessageOrderlyConsumer implements MessageListenerOrderly {
        //消费异常的情况, status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            // 自动提交消费位移
            // context.setAutoCommit(false);
            for (MessageExt message : CollectionUtils.emptyIfNull(msgs)) {
                try {
                    int reconsumeTimes = message.getReconsumeTimes();
                    // 逐条消费
                    String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
                    log.info("Message topic:{} id:{} keys:{} tags:{} body:{} ",message.getTopic() ,message.getMsgId(), message.getKeys(),message.getTags(),messageBody);
                    List<StripCommand> commands = JSON.parseArray(messageBody,StripCommand.class);
                    if(reconsumeTimes>=MQConst.MAX_RECONSUME_TIMES){
                        return ConsumeOrderlyStatus.SUCCESS;
                    }
                    iStripHardwareService.consumeCommands(commands,false);
                    // 模拟耗时操作2分钟，大于设置的消费超时时间
                    // Thread.sleep(1000L * 60 * 2);
                    return ConsumeOrderlyStatus.SUCCESS;
                } catch (Throwable e) {
                    log.error("MQConsumerError:{}", e.getMessage(),e);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
            return ConsumeOrderlyStatus.SUCCESS;
        }
    }

    private String getNameServer(){
        return StrUtil.format( "{}:9876",deviceProperties.getHostIP());
    }
}
